package com.zs.batch.job;

import com.alibaba.fastjson.JSONObject;
import com.zs.batch.listener.JobListener;
import com.zs.batch.listener.SetpListenter;
import com.zs.batch.listener.SkipListener;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.batch.core.BatchStatus;
import org.springframework.batch.core.Job;
import org.springframework.batch.core.Step;
import org.springframework.batch.core.configuration.annotation.EnableBatchProcessing;
import org.springframework.batch.core.configuration.annotation.JobBuilderFactory;
import org.springframework.batch.core.configuration.annotation.StepBuilderFactory;
import org.springframework.batch.core.launch.support.RunIdIncrementer;
import org.springframework.batch.item.ItemProcessor;
import org.springframework.batch.item.ItemReader;
import org.springframework.batch.item.ItemWriter;
import org.springframework.batch.item.file.FlatFileItemReader;
import org.springframework.batch.item.file.LineMapper;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.core.io.FileSystemResource;
import org.springframework.core.task.SimpleAsyncTaskExecutor;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;

import javax.annotation.Resource;
import java.io.File;

/**
 * @author lanyuping
 * @date 2022/11/02 16:39
 */
@Configuration
@EnableBatchProcessing
public class DataJob {

    private static final Logger log = LoggerFactory.getLogger(DataJob.class);

    @Resource
    private JobBuilderFactory jobBuilderFactory;    //用于构建JOB

    @Resource
    private StepBuilderFactory stepBuilderFactory;  //用于构建Step

    @Resource
    private JobListener jobListener;            //简单的JOB listener
    @Resource
    private SetpListenter stepListener;          //简单的Step listener
    @Resource
    private SkipListener skipListener;           //简单的Skip listener


    /**
     * 一个简单基础的Job通常由一个或者多个Step组成
     */
    @Bean
    public Job DataJob() {
        return jobBuilderFactory.get("DataJob").
                incrementer(new RunIdIncrementer()). //设置Job的唯一标识
                start(handleDataStep()).    //start是JOB执行的第一个step
//                next(xxxStep()).
//                next(xxxStep()).
//                ...
                listener(jobListener).      //设置了一个简单JobListener

                        build();
    }

    /**
     * 一个简单基础的Step主要分为三个部分
     * ItemReader : 用于读取数据
     * ItemProcessor : 用于处理数据
     * ItemWriter : 用于写数据
     */
    @Bean
    public Step handleDataStep() {
        return stepBuilderFactory.get("getData").
                <Access, Access>chunk(10).        // <输入,输出> 。chunk通俗的讲类似于SQL的commit; 这里表示处理(processor)100条后写入(writer)一次。
                faultTolerant().retryLimit(3). //容错处理，重试3次
                retry(Exception.class).
                skipLimit(100).skip(Exception.class). //捕捉到异常就重试,重试100次还是异常,JOB就停止并标志失败
                reader(getDataReader()).         //指定ItemReader
                processor(getDataProcessor()).   //指定ItemProcessor
                writer(getDataWriter()).         //指定ItemWriter
                listener(stepListener).
                //taskExecutor(taskExecutor()).      //指定并发执行的线程池
                taskExecutor(new SimpleAsyncTaskExecutor()). //指定并发执行的线程池
                throttleLimit(10).      //并发线程数,默认为4*/
                build();
    }
    String Read= "";

    @Bean
    public ItemReader<? extends Access> getDataReader() {
        //这里可以指定从数据库读取数据，也可以指定从文件读取数据

        FlatFileItemReader<Access> reader=new FlatFileItemReader<>();
        File file = new File("D:\\ideaProject\\batch\\test.txt");
        reader.setResource(new FileSystemResource(file))    ;
        reader.setLineMapper(new LineMapper<Access>() {
            @Override
            public Access mapLine(String line, int lineNumber) {

               // try {
                    log.info("Reader data : {}",line);
                    Access access = new Access();
                    String[] split = line.split(",");
                    access.setId(split[0]);
                    access.setName(split[1]);
                    access.setAge(Integer.valueOf(split[2]));
                    return access;
                /*} catch (Exception e) {
                    log.error("解析数据异常", e);
                }
                return null;*/
            }
        });
        return reader;

    }

    @Bean
    public ItemProcessor<Access, Access> getDataProcessor() {
        return new ItemProcessor<Access, Access>() {
            @Override
            public Access process(Access access) throws Exception {
                log.info("processor data : " + JSONObject.toJSONString(access));  //模拟  假装处理数据,这里处理就是打印一下
                return access;
            }
        };
//        lambda也可以写为:
//        return access -> {
//            log.info("processor data : " + access.toString());
//            return access;
//        };
    }

    @Bean
    public ItemWriter<Access> getDataWriter() {
        return list -> {
            for (Access access : list) {
                log.info("write data : " + JSONObject.toJSONString(access)); //模拟 假装写数据 ,这里写真正写入数据的逻辑
            }
        };
    }


    @Bean
    public ThreadPoolTaskExecutor taskExecutor() {
        ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
        executor.setCorePoolSize(5);//核心线程数
        executor.setMaxPoolSize(20);//设置最大线程数
        executor.setQueueCapacity(100);//队列大小
        executor.setThreadNamePrefix("batch-"); //线程名称前缀
        executor.initialize();//初始化
        return executor;
    }
}
